package com.yungu.swift.socket.loadbalance;


import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;
import com.alibaba.dubbo.rpc.RpcException;
import com.alibaba.dubbo.rpc.cluster.LoadBalance;
import com.yungu.swift.autoconfig.redis.RedisCacheService;
import com.yungu.swift.base.SpringContextHolder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Configuration;

import java.util.List;

/**
 * 长连接集群自定义负载均衡扩展
 * 实现原理：在需要调用PushService服务的模块注册该扩展，当调用PushService中的推送方法时
 * 将会被该扩展拦截，扩展中会获取请求方法中的客户端ID，然后从缓存中通过该客户端ID获取
 * 该客户端连接的长连接服务器IP与端口（连接信息在客户端建立连接时就已经写入缓存中）。
 * 此时该扩展针对连接信息从dubbo注册服务器中注册的所有推送服务中过滤出一个匹配的服务并返回给调用者
 * 这样即巧妙的实现了客户端连接指定长连接服务的方案
 * <p>
 * 注意情况：对于单个推送和批量推送实现的原理有所差异，对于单个推送参考上文；
 * 针对批量推送，情况比较复杂，因为批量推送列表中的信息来自不同客户端，这些客户端可能连接了不同服务器
 * 所以无法按算法返回一个指定服务器，只能将推送列表在每一个服务器都推送一遍
 * 此时对于单台服务器因为部分客户端不存在连接通道将会推送失败，但因为其他服务器推送成功了，所以整体不
 * 影响推送的到达率。
 *
 * @author Zerrion
 * @version v1.3.0
 * @since 2019-5-15 13:52
 */

@Slf4j
@Configuration
public class SocketClusterLoadBalance implements LoadBalance {

    static final String SOCKET_CLUSTER_PREFIX = "YUNGU_REDIS_SOCKET_CLUSTER_";

    @Override
    public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException {
        //判断调用点为PushService服务且方法为sendMessage
        if (url.getServiceInterface().equals("com.yungu.swift.socket.push.service.PushService") && invocation.getMethodName().equals("sendMessage")) {
            //获取调用方法的参数
            Object[] args = invocation.getArguments();
            String targetUuid = null;
            //取targetUuid参数值
            for (Object o : args) {
                if (o instanceof String) {
                    targetUuid = o.toString();
                }
            }
            //取服务列表中与缓存匹配的节点
            for (Invoker invoker : invokers) {
                RedisCacheService cacheService = SpringContextHolder.getBean(RedisCacheService.class);
                Object obj = cacheService.get(SOCKET_CLUSTER_PREFIX + targetUuid, false);
                String socketClusterProvider = obj == null ? "" : obj.toString();
                if (!socketClusterProvider.isEmpty()) {
                    String[] socketClusterProviderArray = socketClusterProvider.split(":");
                    //服务提供者的IP和端口与缓存中该客户端连接的IP及端口比较,如果一致则返回该服务提供者,其他舍弃
                    if (invoker.getUrl().getIp().equals(socketClusterProviderArray[0]) && invoker.getUrl().getPort() == Integer.parseInt(socketClusterProviderArray[1])) {
                        return invoker;
                    } else {
                        continue;
                    }
                }
            }
        } else
            //如果为批量发送，因为在一个批量列表中的客户端可能连接不同长连接服务器，导致无法挑选唯一一台服务器，
            //此处采取一个取巧方式，让集群中各个服务端都将列表中所有消息推送一遍，这样比如一个消息原本连接B服务器
            //在A服务器调用推送方法时就会不能到达，但因为B服务器紧接着也推送了一轮，所以效果一样
            if (url.getServiceInterface().equals("com.yungu.swift.socket.push.service.PushService") && invocation.getMethodName().equals("sendMessageByList")) {
                for (int i = 1; i < invokers.size(); i++) {
                    invokers.get(i).invoke(invocation);
                }
            }
        return invokers.get(0);
    }
}
